package cn.jly.flink.table;

import cn.jly.flink.custom_source.MyStreamingSource;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.ArrayList;

/**
 * 把实时的商品数据流进行分流，分成 even 和 odd 两个流进行 JOIN，条件是名称相同，最后，把两个流的 JOIN 结果输出
 *
 * @author lanyangji
 * @create 2020-09-01 19:23
 */
public class TableApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, environmentSettings);

        DataStreamSource<MyStreamingSource.Item> itemSource = env.addSource(new MyStreamingSource());
        // 分流
        SplitStream<MyStreamingSource.Item> splitStream = itemSource.split((OutputSelector<MyStreamingSource.Item>) value -> {
            ArrayList<String> output = new ArrayList<>();
            if (value.getId() % 2 == 0) {
                output.add("even");
            } else {
                output.add("odd");
            }

            return output;
        });

        DataStream<MyStreamingSource.Item> evenItems = splitStream.select("even");
        DataStream<MyStreamingSource.Item> oddItems = splitStream.select("odd");

        // 注册表
        tableEnv.createTemporaryView("evenTable", evenItems);
        tableEnv.createTemporaryView("oddTable", oddItems);

        // sql查询
        Table table = tableEnv.sqlQuery(
                "select a.id aId, a.name aName, b.id bId, b.name bName from evenTable a join oddTable b on a.name = b.name"
        );
        table.printSchema();

        // 指定格式打印输出
        tableEnv.toRetractStream(table, TypeInformation.of(new TypeHint<Tuple4<Integer, String, Integer, String>>() {}))
                .print();

        env.execute("streaming sql job");
    }
}
